package groupOne.app.DWS;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import groupOne.app.BaseAppOneStream;
import groupOne.bean.PaySucCount;
import groupOne.common.Constant;
import groupOne.util.AtguiguUtil;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisSink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Properties;

/**
 * @Author: LixuanFeng
 * @Date: 2022/9/2 10:11
 * @From: GoodGoodStudyDayDayUp
 *
 *  去重：一天之内同一人可能多次支付，所以状态中维护 末次支付日期
 *        如果当天日期与状态日期不一致 或 状态中为null(首次支付)，则人数记为1
 *    封装成一个POJO类
 *
 *         doris 建表语句：
 *                 drop table if exists dws_user_order_unique_window;
 *                 create table if not exists dws_user_order_unique_window
 *                 (
 *                     `stt`                              DATETIME comment '窗口起始时间',
 *                     `edt`                              DATETIME comment '窗口结束时间',
 *                     `cur_date`                         DATE comment '当天日期',
 *                     `pay_suc_ct`                       BIGINT replace comment '支付成功人数'
 *                 ) engine = olap aggregate key (`stt`, `edt`, `cur_date`)
 *                 comment "用户主题支付成功汇总表"
 *                 partition by range(`cur_date`)()
 *                 distributed by hash(`stt`) buckets 10 properties (
 *                   "replication_num" = "3",
 *                   "dynamic_partition.enable" = "true",
 *                   "dynamic_partition.time_unit" = "DAY",
 *                   "dynamic_partition.start" = "-1",
 *                   "dynamic_partition.end" = "1",
 *                   "dynamic_partition.prefix" = "par",
 *                   "dynamic_partition.buckets" = "10",
 *                   "dynamic_partition.hot_partition_num" = "1"
 *                 );

 */
public class DwsUserPaySucCountWindow_lixuanfeng extends BaseAppOneStream {

    public static void main(String[] args) {

        new DwsUserPaySucCountWindow_lixuanfeng().init(3104,
                2,
                "DwsUserPaySucCountWindow_lixuanfeng",
                Constant.TOPIC_DWD_TRADE_PAY_DETAIL);
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        //1.转换JSON对象格式，并把支付成功的数据过滤出来
        SingleOutputStreamOperator<JSONObject> filterStream = mapFilterData(stream);

        //2.分组，统计结果封装到POJO
        SingleOutputStreamOperator<PaySucCount> pojoStream = countAndPojo(filterStream);

        //3.开窗聚合
        SingleOutputStreamOperator<PaySucCount> aggStream = WindowAgg(pojoStream);
        aggStream.print("PaySucCount");

        //4.写入doris
      //  writeToDoris(aggStream);

    }

    private void writeToDoris(SingleOutputStreamOperator<PaySucCount> aggStream) {

        Properties props = new Properties();
        props.setProperty("format", "json");
        props.setProperty("strip_outer_array", "true");
        String tableName= "edu.dws_user_pay_suc_unique_window";

        aggStream
                .map(pojo -> {
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;  // 转成json的时候, 属性名使用下划线
                    return JSON.toJSONString(pojo, config);
                })
                .addSink(DorisSink.sink(new DorisExecutionOptions.Builder()
                                .setBatchIntervalMs(2000L)
                                .setBatchSize(1024 * 1024)
                                .setEnableDelete(false)
                                .setMaxRetries(3)
                                .setStreamLoadProp(props)
                                .build(),
                        new DorisOptions.Builder()
                                .setFenodes(Constant.DORIS_HOST)
                                .setUsername("root")
                                .setPassword("aaaaaa")
                                .setTableIdentifier(tableName)
                                .build()));



    }

    private SingleOutputStreamOperator<PaySucCount> WindowAgg(SingleOutputStreamOperator<PaySucCount> pojoStream) {
        return
        pojoStream
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5L)))
                .reduce(new ReduceFunction<PaySucCount>() {
                            @Override
                            public PaySucCount reduce(PaySucCount value1, PaySucCount value2) throws Exception {
                                value1.setPaySucCt(value1.getPaySucCt() + value2.getPaySucCt());
                                return value1;
                            }
                        }
                        ,
                        new ProcessAllWindowFunction<PaySucCount, PaySucCount, TimeWindow>() {
                            @Override
                            public void process(Context context, Iterable<PaySucCount> elements, Collector<PaySucCount> out) throws Exception {
                                PaySucCount data = elements.iterator().next();
                                String start = AtguiguUtil.toDate(context.window().getStart());
                                String end = AtguiguUtil.toDate(context.window().getEnd());
                                String curDate = AtguiguUtil.toDate(System.currentTimeMillis());

                                data.setStt(start);
                                data.setEdt(end);
                                data.setCurDate(curDate);

                                out.collect(data);
                            }
                        }
                );
    }

    private SingleOutputStreamOperator<PaySucCount> countAndPojo(SingleOutputStreamOperator<JSONObject> filterStream) {
        return
        filterStream
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5L))
                         .withTimestampAssigner((json,ts)->json.getLong("ts")*1000))
                .keyBy(json->json.getString("user_id"))
                .process(new KeyedProcessFunction<String, JSONObject, PaySucCount>() {

                    private ValueState<String> lastPayDate;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastPayDate = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastPayDate", String.class));
                    }

                    @Override
                    public void processElement(JSONObject value, Context ctx, Collector<PaySucCount> out) throws Exception {
                        String lastPay = lastPayDate.value();
                        Long ts = value.getLong("ts");
                        String thisPay = AtguiguUtil.toDate(ts);

                        Long paySucCt=0L;

                        if (lastPay==null || !thisPay.equals(lastPay)){
                            paySucCt=1L;
                            lastPayDate.update(thisPay);
                        }

                        out.collect( PaySucCount.builder()
                                .paySucCt(paySucCt)
                                .build() );
                    }
                });


    }

    private SingleOutputStreamOperator<JSONObject> mapFilterData(DataStreamSource<String> stream) {
        return
        stream
                .map(jsonStr-> JSON.parseObject(jsonStr))
                .filter(new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        String callbackTime = value.getString("callback_time");
                        String paymentStatus = value.getString("payment_status");
                        return callbackTime!=null && "1602".equals(paymentStatus);
                    }
                });
    }
}
